package com.chukun.flink.stream.state.operator;

import com.chukun.flink.stream.state.key.KeyStateBase;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Calendar;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * @version 1.0.0
 * @description 实现ListCheckpointed接口，使得该函数具备Operator状态
 * @createTime 2022年05月16日 18:08:00
 */
public class ListCheckpointedMapOperator implements MapFunction<Long, String>, ListCheckpointed<Long> {

    private static final Logger logger = LoggerFactory.getLogger(KeyStateBase.StateSource.class);

    // 集合中存储本该输出到外部系统的数据
    private List<Long> bufferedElements;

    // 是否让函数产生异常
    private boolean isError;

    public ListCheckpointedMapOperator(boolean isError) {
        this.isError = isError;
        this.bufferedElements = new LinkedList<>();
    }

    @Override
    public String map(Long value) throws Exception {
        int size = bufferedElements.size();
        if (size >= 10) {
            for (int i = 0; i < size - 9; i++) {
                bufferedElements.remove(0);
            }
        }
        bufferedElements.add(value);
        if (isError) {
            int seconds = Calendar.getInstance().get(Calendar.SECOND);
            if (seconds >= 50 && seconds <= 51) {
                int i = 1 / 0;
            }
        }
        logger.info("{} map data :{}", Thread.currentThread().getName(), bufferedElements);
        return "集合中第一个元素是:" + bufferedElements.get(0) +
                "集合中最后一个元素是:" + bufferedElements.get(bufferedElements.size() -1) +
                " length is :" + bufferedElements.size();
    }

    @Override
    public List<Long> snapshotState(long checkpointId, long timestamp) throws Exception {
        logger.info("{} 快照编号 {} 的元素为 : {}", Thread.currentThread().getName(), checkpointId, bufferedElements);
        return bufferedElements;
    }

    @Override
    public void restoreState(List<Long> state) throws Exception {
        bufferedElements = state;
        logger.info("快照恢复 {}  当前的数据 : {}", Thread.currentThread().getName(), bufferedElements);
    }


    /**
     * 定义数据源
     */
    public static class CustomSource extends RichSourceFunction<Long> {

        public static final Logger logger = LoggerFactory.getLogger(CustomSource.class);

        private static final long serialVersionUID = 1L;

        /**
         * @date: 2022/05/15 18:33
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            TimeUnit.SECONDS.sleep(10);
        }

        /**
         * @date: 2022/05/15 18:33
         */
        @Override
        public void run(SourceContext<Long> ctx) throws Exception {
            Long offset = 0L;
            while (true) {
                logger.info("{}{}{}", Thread.currentThread().getName(), ":发送数据:", offset);
                ctx.collect(offset);
                offset += 1;
                TimeUnit.SECONDS.sleep(1);
            }
        }

        /**
         */
        @Override
        public void cancel() {}
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置每10秒执行一次检查点操作
        env.enableCheckpointing(10000);

        // 设置作业的全局并行度为2
        env.setParallelism(2);

        String path = "file:///E:/checkoutpoint/ListCheckpointedMapOperator";
        // 设置FsStateBackend作为状态后端，检查点文件存储在指定的路径
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(path);

        // 设置作业失败的重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.of(5, TimeUnit.MINUTES),Time.of(10, TimeUnit.SECONDS)));

        DataStream<Long> streamSource = env.addSource(new CustomSource()).setParallelism(1);

        DataStream<String> mapResult = streamSource.map(new ListCheckpointedMapOperator(true));

        mapResult.print("输出结果");

        env.execute("ListCheckpointedMapOperator");

    }
}
